package org.activemonitoring.harvesters;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import org.activemonitoring.datastore.DataStore;
import org.activemonitoring.domain.Measure;
import org.activemonitoring.domain.Measures;
import org.apache.commons.lang.builder.ToStringBuilder;
import org.apache.log4j.Logger;
import org.aspectj.lang.JoinPoint;

import edu.emory.mathcs.backport.java.util.concurrent.Executors;

//@Component
public class HarvesterImpl implements Harvester {

    private static final int DEFAULT_RETENTION_COUNT = 10;
    private static final int DEFAULT_TIMEOUT = 5;
    private static final TimeUnit DEFAULT_TIMEUNIT = TimeUnit.SECONDS;

    private static final Logger logger = Logger.getLogger(HarvesterImpl.class);

    private DataStore dataStore;
    private final List<Measure> measures;
    private final CyclicLatch latch;
    private HarvesterThread harvesterThread;

    public HarvesterImpl() {
        this(DEFAULT_RETENTION_COUNT, DEFAULT_TIMEOUT, DEFAULT_TIMEUNIT);
    }

    public HarvesterImpl(int elementCount, long timeOut, TimeUnit timeUnit) {
        super();
        this.measures = new LinkedList<Measure>();
        this.latch = createCyclicLatch(elementCount, timeOut, timeUnit);
        this.harvesterThread = createThread(this.measures, this.latch, this.dataStore);
    }

    protected CyclicLatch createCyclicLatch(int elementCount, long timeOut, TimeUnit timeUnit) {
        CyclicLatch cyclicLatch = new CyclicLatch(elementCount, timeOut, timeUnit);
        if (logger.isDebugEnabled()) {
            logger.debug("Creating new CyclicLatch: " + cyclicLatch.toString());
        }
        return cyclicLatch;
    }

    protected HarvesterThread createThread(List<Measure> measures, CyclicLatch latch, DataStore dataStore) {
        HarvesterThread harvesterThread = new HarvesterThread(this.measures, this.latch, this.dataStore);
        Executors.defaultThreadFactory().newThread(harvesterThread).start();
        return harvesterThread;
    }

    @Override
    public MeasureBuilder start(String moduleName, JoinPoint joinPoint) {
        return new MeasureBuilderImpl(this, moduleName, createName(joinPoint));
    }

    @Override
    public MeasureBuilder start(String moduleName, String name) {
        logger.debug("Starting a new measure");
        return new MeasureBuilderImpl(this, moduleName, name);
    }

    protected String createName(JoinPoint joinPoint) {
        String name = joinPoint.getTarget().getClass().getName() + '.' + joinPoint.getSignature().getName();
        return name;
    }

    @Override
    public void harvest(Measure measure) {
        logger.debug("Received one complete measure");
        synchronized (this.measures) {
            this.measures.add(measure);
        }
        this.latch.countDown();
    }

    public void setElementCount(int elementCount) {
        this.latch.setElementCount(elementCount);
    }

    public void setTimeout(long timeout, TimeUnit timeUnit) {
        this.latch.setTimeout(timeout, timeUnit);
    }

    public void setDataStore(DataStore dataStore) {
        this.dataStore = dataStore;
        this.harvesterThread.setDataStore(dataStore);
    }

    public DataStore getDataStore() {
        return this.dataStore;
    }

    public List<Measure> getMeasures() {
        return this.measures;
    }

    @Override
    public String toString() {
        return ToStringBuilder.reflectionToString(this);
    }

    protected static class HarvesterThread implements Runnable {

        private DataStore dataStore;
        private List<Measure> measures;
        private CyclicLatch latch;
        private boolean running = true;

        public HarvesterThread(List<Measure> measures, CyclicLatch latch, DataStore dataStore) {
            super();
            this.measures = measures;
            this.latch = latch;
            this.dataStore = dataStore;
        }

        @Override
        public void run() {
            try {
                do {
                    // Wait until list contains enough elements or time is elapsed
                    this.latch.await();
                    if (this.measures.size() != 0) {
                        Measures measureList = new Measures();
                        synchronized (this.measures) {
                            measureList.getMeasure().addAll(this.measures);
                            this.measures.clear();
                        }
                        this.dataStore.store(measureList);
                    }
                } while (this.running);
            } catch (InterruptedException e) {
                logger.warn("AsyncSender thread has been interrupted");
            }
        }

        public void setDataStore(DataStore dataStore) {
            this.dataStore = dataStore;
        }

        public boolean isRunning() {
            return this.running;
        }

        public void setRunning(boolean running) {
            this.running = running;
        }
    }
}
